package net.qiqbframework.eventhandling;

import lombok.extern.slf4j.Slf4j;
import net.qiqbframework.common.Registration;
import net.qiqbframework.messaging.DefaultInterceptorChain;
import net.qiqbframework.messaging.MessageHandlerInterceptor;
import net.qiqbframework.messaging.ResultMessage;
import net.qiqbframework.messaging.monitoring.MessageMonitor;
import net.qiqbframework.messaging.monitoring.MonitorCallback;
import net.qiqbframework.messaging.task.MessageTask;
import net.qiqbframework.messaging.task.RollbackConfiguration;

import javax.annotation.Nonnull;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
public abstract class AbstractEventProcessor implements EventProcessor {

    private final String name;
    /**
     * 事件执行器。相同组事件将合并
     */
    private final EventMessageHandler eventHandlerInvoker;
    private final RollbackConfiguration rollbackConfiguration;
    private final ErrorHandler errorHandler;
    private final MessageMonitor<? super EventMessage<?>> messageMonitor;
    private final List<MessageHandlerInterceptor<? super EventMessage<?>>> interceptors = new CopyOnWriteArrayList<>();
    protected final EventProcessorSpanFactory spanFactory;


    protected AbstractEventProcessor(AbstractEventProcessorBuilder builder) {
        builder.validate();
        this.name = builder.name;
        this.eventHandlerInvoker = builder.getEventHandlerInvoker();
        this.rollbackConfiguration = builder.getRollbackConfiguration();
        this.errorHandler = builder.getErrorHandler();
        this.messageMonitor = builder.getMessageMonitor();
        this.spanFactory = builder.getSpanFactory();
    }

    @Override
    public String getName() {
        return name;
    }

    @Override
    public Registration registerHandlerInterceptor(  @Nonnull MessageHandlerInterceptor<? super EventMessage<?>> interceptor) {
        interceptors.add(interceptor);
        return () -> interceptors.remove(interceptor);
    }

    @Override
    public List<MessageHandlerInterceptor<? super EventMessage<?>>> getHandlerInterceptors() {
        return Collections.unmodifiableList(interceptors);
    }

    @Override
    public String toString() {
        return getName();
    }





    protected void processInUnitOfWork(EventMessage<?> eventMessages, MessageTask<? extends EventMessage<?>> unitOfWork) throws Exception {
        ResultMessage<?> resultMessage = unitOfWork.executeWithResult(() -> {
            EventMessage<?> message = unitOfWork.getMessage();
            MonitorCallback monitorCallback = messageMonitor.onMessageIngested(message);
            return spanFactory.createProcessEventSpan(message)
                    .runCallable(() -> new DefaultInterceptorChain<>(
                            unitOfWork,
                            interceptors,
                            m -> processMessageInUnitOfWork(m, monitorCallback))
                            .proceed());
        }, rollbackConfiguration);

        if (resultMessage.isExceptional()) {
            Throwable e = resultMessage.exceptionResult();
            if (unitOfWork.isRolledBack()) {
                errorHandler.handleError(new ErrorContext(getName(), e, eventMessages));
            } else {
                log.info(
                        "Exception occurred while processing a message, but unit of work was committed. {}",
                        e.getClass().getName());
            }
        }
    }

    private Object processMessageInUnitOfWork(EventMessage<?> message, MonitorCallback monitorCallback) throws Exception {
        try {
            eventHandlerInvoker.handle(message);
            monitorCallback.reportSuccess();
            return null;
        } catch (Exception exception) {
            monitorCallback.reportFailure(exception);
            throw exception;
        }
    }


    protected void reportIgnored(EventMessage<?> eventMessage) {
        messageMonitor.onMessageIngested(eventMessage).reportIgnored();
    }


}
